package com.df.my_net_game.net.base;

import com.baidu.bjf.remoting.protobuf.Codec;
import com.baidu.bjf.remoting.protobuf.ProtobufProxy;
import com.df.my_net_game.common.utils.LogicMethodData;
import com.df.my_net_game.common.utils.LogicMethodMgr;
import com.df.my_net_game.game_rpc.*;
import com.df.my_net_game.game_rpc.types.ByteWrapper;
import com.df.my_net_game.game_rpc.types.IntWrapper;
import com.df.my_net_game.game_rpc.types.LongWrapper;
import com.df.my_net_game.game_rpc.types.ShortWrapper;
import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.util.*;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
public class EzGameMsgDispatcher {


    private static final EzGameMsgDispatcher instance = new EzGameMsgDispatcher();
    //扫描路径
    List<Disruptor<EzGameMsgWrapper>> msgQueue = new ArrayList<Disruptor<EzGameMsgWrapper>>();

    Set<IMsgInterceptor> interceptors = new HashSet<>();


    @Getter
    @Setter
    IMsgGroup msgGroup;

    private EzGameMsgDispatcher(){

        for (int i = 0; i < 16; i++) {
            int index = i;
            Disruptor<EzGameMsgWrapper> disruptor = new Disruptor<EzGameMsgWrapper>(new MsgEventFactory(),
                    2<<16,
                    (Runnable r)->new Thread(r,"msg_handle_thread_"+index),
                    ProducerType.MULTI,
                    new TimeoutBlockingWaitStrategy(0,TimeUnit.SECONDS)
            );
            disruptor.handleEventsWith(new MsgEventHandler());
            disruptor.start();
            msgQueue.add(disruptor);
        }

    }

    public static EzGameMsgDispatcher getInstance(){
        return instance;
    }

    public int getQueueSize() {
        return msgQueue.size();
    }

    public void addInterceptor(IMsgInterceptor interceptor) {
        interceptors.add(interceptor);
    }

    public void dispatcherMsg(EzGameMsg realMsg, ChannelHandlerContext ctx) {
        int nextQueueIndex = getNextQueueIndex(realMsg,ctx);
        Disruptor<EzGameMsgWrapper> disruptor = msgQueue.get(nextQueueIndex);
        //尝试获取，获取不到，则直接丢弃消息，给客户端返回繁忙
        long cursor = 0;
        EzGameMsgWrapper msg = null;
        try {
            RingBuffer<EzGameMsgWrapper> ringBuffer = disruptor.getRingBuffer();
            cursor = disruptor.getRingBuffer().tryNext();
            msg = disruptor.get(cursor);
            msg.setMsg(realMsg);
            ringBuffer.publish(cursor);
        } catch (InsufficientCapacityException e) {
            //TODO 有这个异常，表示生产过快，队列满了，处理太慢，返回繁忙的消息，直接给前端返回
            if(msg!=null){//防止环对象持有消息，造成内存泄露
                msg.setMsg(null);
            }
            log.error("队列已满，放弃消息！请检查是否业务方法执行时间过长！");
        }catch (Exception e){
            e.printStackTrace();
            if(msg!=null){//防止环对象持有消息，造成内存泄露
                msg.setMsg(null);
            }
        }
    }


    private static class MsgEventHandler implements EventHandler<EzGameMsgWrapper>{

//        EzGameMsgDispatcher dispatcher;
//        Set<IMsgInterceptor> interceptors;

        public MsgEventHandler() {
        }

        @Override
        public void onEvent(EzGameMsgWrapper msg, long l, boolean b) throws Exception {
            EzGameMsg reqMsg = msg.getMsg();
            msg.setMsg(null);//移除掉被包装对象，防止内存泄露
            if(reqMsg==null){
                return;
            }
            LogicMethodData logicMethodData = LogicMethodMgr.getLogicMethodData(reqMsg.getHeader().getCmdId());
            if(logicMethodData==null){
                return;
            }
            //TODO 缓存起来，防止频繁创建对象，使用引用计数
            EZGameCtx context = new EZGameCtx();
            context.setCmdId(reqMsg.getHeader().getCmdId());
//            context.setReqId(reqMsg.getHeader().getReqId());
            context.setReqMsg(reqMsg);

            context.setChannel(reqMsg.getChannel());
            EZGameCtx.setGameCtx(context);


//            for (IMsgInterceptor interceptor : interceptors) {
//                try {
//                    interceptor.beforeInvoke(context);
//                }catch (Exception e){
//                    e.printStackTrace();
//                }
//            }

            try {
//                        ValidationUitl
                //TODO 数据校验
                Object param = null;
                if(logicMethodData.getParamType()!=null){
                    if(reqMsg.getBodyBytes()==null || reqMsg.getBodyBytes().length==0){
                        throw new RuntimeException("消息请求体为空！消息id："+reqMsg.getCmdId());
                    }
                    if(logicMethodData.isParamIsBaseType()){
                        Class paramType = logicMethodData.getParamWrapperType();
                        Codec codec = ProtobufProxy.create(paramType);
                        Object decodeParam = codec.decode(reqMsg.getBodyBytes());
                        param = getRawParam(decodeParam,paramType);
                    }else {
                        Class paramType = logicMethodData.getParamType();
                        Codec codec = ProtobufProxy.create(paramType);
                        param = codec.decode(reqMsg.getBodyBytes());
                    }
                }

                Object resp = logicMethodData.invoke(param);
                Class returnType = logicMethodData.getReturnType();
                if(returnType!=null){//需要有返回值
                    if(resp!=null){//TODO 检查是否要给客户端发消息
                        EzGameMsg rpcResp = new EzGameMsg();
                        context.setRespMsg(rpcResp);

                        rpcResp.setBodyObj(resp);
                        rpcResp.setChannel(reqMsg.getChannel());

                        GRpcHeader respHeader = new GRpcHeader();
                        respHeader.setReqFrom(1);
                        respHeader.setMsgType(GRpcMsgTypeEnum.Resp.msgType());
                        respHeader.setCmdId(reqMsg.getHeader().getCmdId());
                        rpcResp.setHeader(respHeader);
                        if(logicMethodData.isReturnIsBaseType()){
                            resp = getRawRetunObj(logicMethodData.getReturnWrapperType(),resp);
                            Codec codec = ProtobufProxy.create(logicMethodData.getReturnWrapperType());
                            byte[] respbytes = codec.encode(resp);
                            rpcResp.setBodyBytes(respbytes);
                        }else {
                            Codec codec = ProtobufProxy.create(returnType);
                            byte[] respbytes = codec.encode(resp);
                            rpcResp.setBodyBytes(respbytes);
                        }
                        Channel channel = context.getChannel();
                        if(channel.isOpen() && channel.isActive()){
                            channel.writeAndFlush(rpcResp);
                        }
                    }else{
                        log.warn("请求需要返回值，但返回值为null!cmd:"+reqMsg.getCmdId());
                    }
                }

            }catch (Exception e){
                //TODO 处理全局错误
//                for (IMsgInterceptor interceptor : interceptors) {
//                    try {
//                        interceptor.onException(e,context);
//                    }catch (Exception e1){
//                        e1.printStackTrace();
//                    }
//                }
//                if(logicMethodData.getReturnType())
                if(e instanceof EzGameException){
                    //直接返回
                }else{
                    e.printStackTrace();
                }

            }finally {
//                for (IMsgInterceptor interceptor : interceptors) {
//                    try {
//                        interceptor.afterInvoke(context);
//                    }catch (Exception e){
//                        e.printStackTrace();
//                    }
//                }

                if(context.getRespMsg()!=null){
                    if(context.getChannel()!=null){
//                        if(server.getServerCB()!=null){
//
//                            try {
//                                EzGameMsg respMsg = server.getServerCB().onBeforeSendMsg(context.getRespMsg(), server);
//                                context.setRespMsg(respMsg);
//                            }catch (Exception e){
//                                e.printStackTrace();
//                            }
//                            ByteBuf encode = null;
//                            if(encode==null){
//                                Integer msgType = reqMsg.getHeader().getMsgType();
//                                if(msgType.equals(GRpcMsgTypeEnum.Req.msgType())){
//                                    log.error("客户端 请求返回值编码结果为null，返回消息失败，请检查serverCb的encode方法是否正确！cmdId:"+reqMsg.getHeader().getCmdId());
//                                }else {
//                                    log.error("rpc 请求返回值编码结果为null，返回消息失败，请检查serverCb的encode方法是否正确！cmdId:"+reqMsg.getHeader().getCmdId());
//                                }
//                            }else {
//                                context.getChannel().writeAndFlush(encode);
//                            }
//                        }

                    }
                }
                EZGameCtx.removeContext();
            }
        }

        private Object getRawRetunObj(Class returnWrapperType, Object resp) {
            if(returnWrapperType.equals(ByteWrapper.class)){
                ByteWrapper wrapper = new ByteWrapper();
                wrapper.setValue((Byte) resp);
                return wrapper;
            }
            if(returnWrapperType.equals(ShortWrapper.class)){
                ShortWrapper wrapper = new ShortWrapper();
                wrapper.setValue((Short) resp);
                return wrapper;
            }
            if(returnWrapperType.equals(IntWrapper.class)){
                IntWrapper wrapper = new IntWrapper();
                wrapper.setValue((Integer) resp);
                return wrapper;
            }
            if(returnWrapperType.equals(LongWrapper.class)){
                LongWrapper wrapper = new LongWrapper();
                wrapper.setValue((Long) resp);
                return wrapper;
            }
            throw new RuntimeException("未支持的返回值类型："+returnWrapperType.getName());
        }

        private Object getRawParam(Object decodeParam, Class paramType) {
            if(paramType.equals(ByteWrapper.class)){
                ByteWrapper bw = (ByteWrapper) decodeParam;
                return bw.getValue();
            }
            if(paramType.equals(ShortWrapper.class)){
                ShortWrapper bw = (ShortWrapper) decodeParam;
                return bw.getValue();
            }
            if(paramType.equals(IntWrapper.class)){
                IntWrapper bw = (IntWrapper) decodeParam;
                return bw.getValue();
            }
            if(paramType.equals(LongWrapper.class)){
                LongWrapper bw = (LongWrapper) decodeParam;
                return bw.getValue();
            }
            throw new RuntimeException("未支持的参数类型："+paramType.getName());
        }
    }


    private int getNextQueueIndex(EzGameMsg readMsg, ChannelHandlerContext ctx){
        int group = -1;
        if(msgGroup!=null){
            msgGroup.goupId(readMsg,ctx,msgQueue.size());
        }
        if(group>=msgQueue.size()){
            log.error("当前分配的groupId大于等于组的数量，随机组进行处理");
            group = -1;
        }
        if(group==-1){
            Random random = new Random();
            group = random.nextInt(0, msgQueue.size());
        }
        return group;
    }






    private static class MsgEventFactory implements EventFactory<EzGameMsgWrapper> {

        @Override
        public EzGameMsgWrapper newInstance() {
            return new EzGameMsgWrapper();
        }
    }
    private static class TestLongWrapper{
        public long value;
    }

    private static class MsgEventFactory2 implements EventFactory<EzGameMsgDispatcher.TestLongWrapper> {

        @Override
        public EzGameMsgDispatcher.TestLongWrapper newInstance() {
            return new EzGameMsgDispatcher.TestLongWrapper();
        }
    }
    public static void main(String[] args) {
        AtomicInteger Num = new AtomicInteger(0);
        Disruptor<EzGameMsgDispatcher.TestLongWrapper> disruptor = new Disruptor<EzGameMsgDispatcher.TestLongWrapper>(new MsgEventFactory2(),
                4,
                (Runnable r)->new Thread(r,"msg_handle_thread-"+Num.incrementAndGet()),
                ProducerType.MULTI,
                new TimeoutBlockingWaitStrategy(0, TimeUnit.SECONDS)
        );
        disruptor.handleEventsWith(new EventHandler<EzGameMsgDispatcher.TestLongWrapper>() {
            @Override
            public void onEvent(EzGameMsgDispatcher.TestLongWrapper wrapper, long l, boolean b) throws Exception {
                System.out.println(Thread.currentThread().getName()+":"+wrapper.value+"   hash:"+System.identityHashCode(wrapper));
//                Thread.sleep(1000);
            }
        });

        disruptor.start();
        CountDownLatch count = new CountDownLatch(11);
        for (int i = 0; i < 1; i++) {
            int index = i;
            Thread t = new Thread(()->{
                RingBuffer<EzGameMsgDispatcher.TestLongWrapper> ringBuffer = disruptor.getRingBuffer();
                for (int j = 0; j < 10; j++) {
                    long next = 0;
                    try {
                        next = ringBuffer.tryNext();
                        EzGameMsgDispatcher.TestLongWrapper msg = ringBuffer.get(next);
//                        msg.value = j;
                        msg.value = index*10+j;
                        ringBuffer.publish(next);
                        try {
                            Thread.sleep(100);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    } catch (InsufficientCapacityException e) {
                        e.printStackTrace();
                    }catch (Exception e){
                        System.out.println("其他异常");
                        e.printStackTrace();
                    }

                }
                count.countDown();
                System.out.println(Thread.currentThread().getName()+":生产完成！");
            });
            t.setName("producer_"+i);
            t.start();
        }
        try {
            count.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        disruptor.halt();
    }

}
